/*
 * gy_enpctrl.cpp
 *
 *  Created on: 2015年9月15日
 *      Author: gaoyi
 */


#include "gy_enpctrl.h"
#include "gy_enpdef.h"
#include "gy_enpclient.h"
#include "gy_enpclientfactory.h"


void GYEnpCtrl::HandleShmQueueMsg()
{
	int iRet = 0;
	for(;;)
	{
		if (!m_stShmQueue.HasCode())
		{
			return;
		}

        // 接收消息

		int iCodeLen = 0;
		m_stTempBuf.Clear();
		iRet = m_stShmQueue.GetOneCode(m_stTempBuf.GetFreeBuf(), m_stTempBuf.GetFreeLen(), iCodeLen);
		if (iRet || iCodeLen < (int)sizeof(CEnpNetHead))
		{
			CT_WARNING(("get code from shmqueue failed ret=%d, codelen=%d", iRet, iCodeLen));
			continue;
		}

		m_stTempBuf.Append(iCodeLen);

		// 先获取NetHead
		CEnpNetHead& stHead = *(CEnpNetHead *)m_stTempBuf.GetUsedBuf();
		if (iCodeLen != (int)sizeof(CEnpNetHead)+stHead.unDataLength)
		{
			CT_WARNING(("code length invalid. %d != %d + %d", iCodeLen,
			                sizeof(CEnpNetHead), stHead.unDataLength));
			continue;
		}

		// 寻找对应的Client
		GYEnpClient* pstClient = m_stClientFactory.Find(stHead.uiHandleSeq);
		if (pstClient == NULL)
		{
			CT_WARNING(("nethead not match client: uiHandlerSeq=%u", stHead.uiHandleSeq));
			continue;
		}
		// 检查数据长度是否合法
		if ((int)stHead.unDataLength > GY_ENP_CLIENT_BUFFER_SIZE)
		{
			CT_WARNING(("nethead(%d) datalen(%d > %d) invalid", stHead.uiHandleSeq,
			                stHead.unDataLength, GY_ENP_CLIENT_BUFFER_SIZE));
			continue;
		}

		// 检查IP和端口是否匹配
		if (pstClient->m_stNetHead.uiRemoteIP != stHead.uiRemoteIP ||
				pstClient->m_stNetHead.unRemotePort != stHead.unRemotePort || stHead.uiRemoteIP == 0 || stHead.unRemotePort == 0)
		{
			CT_WARNING(("nethead(%d) ip invalid (%d:%d != %d:%d || %d:%d == 0)",
							stHead.uiHandleSeq,
			                pstClient->m_stNetHead.uiRemoteIP, pstClient->m_stNetHead.unRemotePort,
			                stHead.uiRemoteIP, stHead.unRemotePort,
			                stHead.uiRemoteIP, stHead.unRemotePort));
			continue;
		}

		// 获取上层的flag, 只有当高8位不为0时才设置
		if (stHead.unLiveFlag & 0xFF00)
		{
			pstClient->m_stNetHead.unLiveFlag = stHead.unLiveFlag & 0xFF00;
		}

		// 检查上层状态
		if (pstClient->m_stNetHead.unLiveFlag & enp_flag_busy)
		{
			//
		}

        pstClient->m_stNetHead.u64Act1 = stHead.u64Act1;
        pstClient->m_stNetHead.u64Act2 = stHead.u64Act2;
		memcpy(pstClient->m_stNetHead.szPtKey, stHead.szPtKey, sizeof(pstClient->m_stNetHead.szPtKey));

		pstClient->m_stSendBuf.Clear();

		if (stHead.unDataLength > 0)
		{
			if (pstClient->m_pstListen->m_stListenInfo.m_ucEncodeMethod == CODE_PROTOBUF ||
					pstClient->m_pstListen->m_stListenInfo.m_ucEncodeMethod == CODE_BIN ||
					pstClient->m_pstListen->m_stListenInfo.m_ucEncodeMethod == CODE_JSON)
			{
				int iNeedSendBufLen = (int)(sizeof(int)+stHead.unDataLength);

				if (pstClient->m_stSendBuf.GetFreeLen() < iNeedSendBufLen)
				{
					CT_WARNING(("send buff is not enough"));
					continue;
				}
				//增加四个字节的长度
				int iLen = htonl((int)(stHead.unDataLength+sizeof(int)));

				pstClient->m_stSendBuf.Append((char*)&iLen, sizeof(int));
			}

			iRet = pstClient->m_stSendBuf.Append(m_stTempBuf.GetUsedBuf()+sizeof(CEnpNetHead), stHead.unDataLength);

			if (iRet)
			{
                if (pstClient->m_stSendBuf.GetUsedLen() == 0)
                {
                	pstClient->m_stSendBuf.Clear();
                }

                CT_ERROR(("append data failed, ret=%d, datalen=%d, freelen=%d", iRet, stHead.unDataLength,
                     m_stTempBuf.GetFreeLen()));
				continue;
			}

			CT_TRACE(("handle(%d) has %d bytes to send", stHead.uiHandleSeq, pstClient->m_stSendBuf.GetUsedLen()));
		}

		if (pstClient->m_stSendBuf.GetUsedLen() == 0)
		{
			pstClient->m_stSendBuf.Clear();
			continue;
		}

		iRet = SendSocket(pstClient);
		if (iRet)
		{
			CT_ERROR(("send data failed %d", iRet));
			continue;
		}
	}
}

int GYEnpCtrl::PutOneNotify(GYEnpClient* pstClient, unsigned short unLiveFlag)
{
	if (pstClient == NULL)
	{
		return -1;
	}
	return PutOneNotify(pstClient->m_stNetHead, unLiveFlag);
}

int GYEnpCtrl::PutOneNotify(CEnpNetHead& stNetHead, unsigned short unLiveFlag)
{
    int iRet = 0;

    stNetHead.unLiveFlag = unLiveFlag;
    stNetHead.unDataLength = 0;

    iRet = m_stShmQueue.PutOneCode((char*)(&stNetHead), sizeof(stNetHead));
    if (iRet)
    {
        CT_ERROR(("PutOneNotify failed %d", iRet));
    }

    return iRet;
}

int	GYEnpCtrl::CloseSocket(GYEnpClient* pstClient, unsigned short unLiveFlag)
{
	if (pstClient == NULL)
	{
		return -1;
	}

    // 满足3个条件才通知上层
    // 1.socket有效
    // 2.不是enpsvr主动关闭的
    // 3.连接建立后向上层成功发过数据
	if (pstClient->m_stSocket.IsValid() && !(unLiveFlag & enp_flag_askclose) && pstClient->m_iRecvBytes > 0)
	{
		PutOneNotify(pstClient, unLiveFlag);
	}

	m_stClientFactory.Free(pstClient);
	return 0;
}

int GYEnpCtrl::SendSocket(GYEnpClient* pstClient)
{
	if (!pstClient->m_stSendBuf.IsAttached())
	{
		return 0;
	}

	if (!pstClient->m_stSendBuf.GetFreeLen() <= 0)
	{
		CT_TRACE(("no data to send"));
		return 0;
	}

	int iRet = 0;
	int iSendLen = 0;

    // 根据listen端口的属性来发送消息完整性标识
    // protobuf类型就是先发4字节的长度并保证最后以\t结尾
    // binary类型就是先发4 字节的长度
    // string类型就是保证最后以\n结尾
    // flash类型就是保证最后以\0结尾

	if (pstClient->m_pstListen->m_stListenInfo.m_stPackageType == enp_package_http)
	{
		pstClient->m_stNetHead.unLiveFlag |= enp_flag_askclose;
	}
	else if (pstClient->m_pstListen->m_stListenInfo.m_stPackageType == enp_package_protobuf)
	{
		if (pstClient->m_stSendBuf.GetUsedBuf()[pstClient->m_stSendBuf.GetUsedLen()-1] != '\t')
		{
			if (pstClient->m_stSendBuf.GetFreeLen() == 0)
			{
				CT_TRACE(("handle want to send protobuf msg, need add \\t, but no free space"));
				CloseSocket(pstClient, enp_flag_error);
				return -1;
			}
			pstClient->m_stSendBuf.Append("\t", 1);
		}
	}

	iRet = pstClient->m_stSocket.SendEx(pstClient->m_stSendBuf.GetUsedBuf(), pstClient->m_stSendBuf.GetUsedLen(), iSendLen);
	if (iRet && !(iRet == CT_EWOULDBLOCK || iRet == CT_EINPROGRESS))
	{
		if (CT_ECONNRESET == iRet || EPIPE == iRet)
		{
			CT_WARNING(("handle(%d) write msg size failed, client disconnected", pstClient->m_stNetHead.uiHandleSeq));
		}
		else
		{
			CT_ERROR(("handle(%d) write failed socket=%d err=%d", pstClient->m_stNetHead.uiHandleSeq, pstClient->m_stSocket.GetSocket(), iRet));
		}
		CloseSocket(pstClient, enp_flag_error);
		return iRet;
	}

	if (iSendLen == 0)
	{
        // 可能系统的发送缓冲区已满, 那就先缓存一下, 稍后发送
        CT_WARNING(("handle(%d) send 0 bytes", pstClient->m_stNetHead.uiHandleSeq));
		return 0;
	}

	pstClient->m_iSendBytes += iSendLen;
	pstClient->m_stSendBuf.Remove(iSendLen);
	return AfterSendSocket(pstClient);
}

int GYEnpCtrl::RecvSocket(GYEnpClient* pstClient)
{
	if (pstClient == NULL)
	{
		return -1;
	}

	int iRet = 0;

	pstClient->m_stNetHead.uiLastTime = m_uiNow;
	if (pstClient->m_stRecvBuf.GetFreeLen() <= 0)
	{
		CT_WARNING(("handle(%d) recv buffer no space to recv", pstClient->m_uiSeq));
		return 0;
	}

	int iRecvLen = 0;
	iRet = pstClient->m_stSocket.Recv(pstClient->m_stRecvBuf.GetFreeBuf(), pstClient->m_stRecvBuf.GetFreeLen(), iRecvLen);

	if (iRecvLen > 0)
	{
		pstClient->m_stRecvBuf.Append(iRecvLen);
	}

	CT_TRACE(("recv len = %d, ret = %d", iRecvLen, iRet));

	if (iRet)
	{
		if (iRet == CT_ECONNRESET) // 远程连接已关闭
		{
            AfterRecvSocket(pstClient);  //可能会有notify信息
            return CloseSocket(pstClient, enp_flag_remoteclose);
		}
		else if (iRet == CT_EWOULDBLOCK || iRet == CT_EINPROGRESS)
		{
			return AfterRecvSocket(pstClient);
		}
		else
		{
			CT_ERROR(("handle(%d) recv failed %d", pstClient->m_uiSeq, iRet));
            CloseSocket(pstClient, enp_flag_error);
            return -1;
		}
	}

	return AfterRecvSocket(pstClient);
}

int GYEnpCtrl::BufferHasMessage(GYEnpClient* pstClient, int& iLen, char*& pszMsgBuf, int& iMsgLen)
{
	iLen = 0;
	iMsgLen = 0;
	pszMsgBuf = NULL;

	if (pstClient->m_pstListen->m_stListenInfo.m_stPackageType == enp_package_protobuf)
	{
		if (pstClient->m_stRecvBuf.GetUsedLen() < (int)sizeof(int))
		{
			iMsgLen = 0;
			return 0;
		}

		int iPackageLen = ntohl(*(int*)(pstClient->m_stRecvBuf.GetUsedBuf()));
		if (iPackageLen <= pstClient->m_stRecvBuf.GetUsedLen())
		{
			iLen = iPackageLen;
			pszMsgBuf = pstClient->m_stRecvBuf.GetUsedBuf() + sizeof(int);
			iMsgLen = iLen - sizeof(int);
			if (iMsgLen <= 0)
			{
				CT_ERROR(("Invalid package length (%d) and message length (%d)", iLen, iMsgLen));
				return -2;
			}

			char p = *(char*)(pstClient->m_stRecvBuf.GetUsedBuf()+iLen-1);
			if (p != '\t')
			{
				CT_ERROR(("Invalid package content (len=%d) flag error (%d)", iLen, p));
				return -2;
			}
		}
		else
		{
			iMsgLen = 0;
			return 0;
		}

		if (iMsgLen <= 0)
		{
			return -1;
		}

		return 0;
	}

	return -1;
}

int GYEnpCtrl::AfterSendSocket(GYEnpClient* pstClient)
{
	if (pstClient == NULL)
	{
		return -1;
	}
	CT_TRACE(("---- handle data send finished. left %d bytes ----",
        pstClient->m_stSendBuf.GetUsedLen()));

	if (pstClient->m_stSendBuf.GetUsedLen() == 0)
	{
		if (pstClient->m_stNetHead.unLiveFlag & enp_flag_askclose)
		{
			return CloseSocket(pstClient, enp_flag_askclose);
		}
		else
		{
			pstClient->m_stSendBuf.Clear();
			m_stClientFactory.SetEmpty(pstClient->m_stNetHead.uiHandleSeq);
			return 0;
		}
	}
	else
	{
		return m_stClientFactory.SetRemain(pstClient->m_stNetHead.uiHandleSeq);
	}
}

int GYEnpCtrl::AfterRecvSocket(GYEnpClient* pstClient)
{
	// 判断Buf中是否有消息可以放到接收队列中
	int iLen = 0;
	int	iRet = 0;
	char* szMsgBuf = NULL;
	int	  iMsgLen = 0;

	while(true)
	{
		iRet = 0;
		iMsgLen = 0;
		szMsgBuf = NULL;
		iLen = 0;

		if (pstClient->m_stRecvBuf.GetUsedLen() == 0)
		{
			CT_TRACE(("recvbuf len = 0, PutOneMessage over"));
			break;
		}

		iRet = BufferHasMessage(pstClient, iLen, szMsgBuf, iMsgLen);
		if (iRet || iMsgLen == 0 || szMsgBuf == NULL)
		{
			if (iRet == 0 && iMsgLen != 0 && iMsgLen > pstClient->m_stRecvBuf.GetBufferSize())
			{
                CT_WARNING(("recv a msg len great than recv buff, msglen = %d, bufmax = %d",
					iMsgLen, pstClient->m_stRecvBuf.GetBufferSize()));
				CloseSocket(pstClient, enp_flag_msg_too_big);
				return 0;
			}
			else if (-2 == iRet)
			{
				CT_WARNING(("BufferHasMessage returns %d, close this client instance", iRet));
				CloseSocket(pstClient, enp_flag_error);
				return 0;
			}
			CT_TRACE(("no msg in buf, ret = %d, msglen = %d, buf = %p", iRet, iMsgLen, szMsgBuf));
			break;
		}

		CT_TRACE(("Has one message len=%d msglen=%d, put to RecvQueue", iLen, iMsgLen));
		iRet = PutOneMessage(pstClient, iLen, szMsgBuf, iMsgLen);
		if (iRet)
		{
			CT_ERROR(("PutOneMessage failed %d", iRet));
			break;
		}
	}

	if (pstClient->m_stRecvBuf.GetUsedLen() == 0)
	{
		pstClient->m_stRecvBuf.Clear();
	}
	return iRet;
}

int GYEnpCtrl::PutOneMessage(GYEnpClient* pstClient, int iLen, const char* szMsgBuf, int iMsgLen)
{
	int iRet = 0;

	CT_TRACE(("put one message to RecvQueue"));

	pstClient->m_stNetHead.uiHandleSeq = pstClient->m_uiSeq;
	pstClient->m_stNetHead.unDataLength = static_cast<unsigned short>(iMsgLen);

	iRet = m_stShmQueue.PutOneCode((const char *)(&pstClient->m_stNetHead), sizeof(pstClient->m_stNetHead), szMsgBuf, iMsgLen);
	if (iRet)
	{
		return -3;
	}

	pstClient->m_iRecvBytes += iLen;
	pstClient->m_stRecvBuf.Remove(iLen);

	return 0;
}

void GYEnpCtrl::OnShmQueueEvent(CTEpollObject<GYEnpCtrl>* pstObject, SOCKET iSocket, int iEvent)
{
	m_stShmQueue.ClearNotify();
	if (!CTEpoll::IsInputEvent(iEvent))
	{
		return;
	}
}

void GYEnpCtrl::OnListenEvent(CTEpollObject<GYEnpCtrl>* pstObject, SOCKET iSocket, int iEvent)
{
	 CT_TRACE(("recv listen event, event = %d", iEvent));
	if (!CTEpoll::IsInputEvent(iEvent))
	{
		CT_WARNING(("listen socket recv not match event(%d) socket=%d", iEvent, iSocket));
		return;
	}

	int iRet = 0;
	GYEnpListen* pstListen = (GYEnpListen*)pstObject;
	pstListen->m_uiLastTime = m_uiNow;

	struct sockaddr_in clientaddr;
	socklen_t clilen = sizeof(clientaddr);
	SOCKET s = INVALID_SOCKET;
	iRet = pstListen->m_stSocket.Accept(s, (sockaddr *)&clientaddr, &clilen);
	if (iRet)
	{
        // 客户端连接上来以后又立即关闭了
        CT_WARNING(("remote connect and close. listen=%d ret=%d errno=%d",
            iSocket, iRet, CT_WSA_ERRNO));
		return;
	}

	GYEnpClient	*pstClient = m_stClientFactory.Alloc();
	if (pstClient == NULL)
	{
		CT_WARNING(("m_stClientFactory.Alloc failed"));
		return;
	}

	iRet = pstClient->Init(*this, *pstListen, s, clientaddr.sin_addr.s_addr, ntohs(clientaddr.sin_port));
	CHECK_THROW(iRet);

    CT_TRACE(("Client %s:%d connected port %d at sock %d",
        inet_ntoa(clientaddr.sin_addr),
        ntohs(clientaddr.sin_port),
        pstListen->m_stListenInfo.m_unListenPort,
        s));
}

void GYEnpCtrl::OnClientEvent(CTEpollObject<GYEnpCtrl>* pstObject, SOCKET iSocket, int iEvent)
{
	GYEnpClient* pstClient = (GYEnpClient*)pstObject;

	if(CTEpoll::IsCloseEvent(iEvent))
	{
		CloseSocket(pstClient, enp_flag_remoteclose);
	}

	if(CTEpoll::IsInputEvent(iEvent))
	{
        // 接收数据;
        // 如果是完整的包就放入CTShmQueue;
        RecvSocket(pstClient);
	}

	if(CTEpoll::IsOutputEvent(iEvent))
	{
		return;
	}
}

int	GYEnpCtrl::Init()
{
	int iRet = 0;
	iRet = m_stEpoll.Init(GY_ENP_MAX_SOCKET_SIZE, GY_ENP_WAIT_SIZE);
	CHECK_THROW(iRet);

	iRet = m_stShmQueue.Init(false,GY_ENP_FRONT_END_SHM_KEY,GY_ENP_FRONT_END_SHM_SIZE,GY_ENP_FRONT_END_SOCKET, GY_ENP_BACK_END_SOCKET, m_stEpoll, *this, &GYEnpCtrl::OnShmQueueEvent);
	CHECK_THROW(iRet);

	iRet = m_stListen.Init(*this, GY_ENP_LISTEN_IP, GY_ENP_LISTEN_PORT);
	CHECK_THROW(iRet);

	m_stTempBuf.Attach(NULL, GY_ENP_CLIENT_BUFFER_SIZE+sizeof(CEnpNetHead), 0);

	CT_INFO(("ENP init ok"));
	return 0;
}

void GYEnpCtrl::DumpStatInfo()
{
	CT_STAT->Log(EInfo, "Handle: All=%d Remain=%d",
        m_stClientFactory.GetClientSize(), m_stClientFactory.GetRemainSize());
}

void GYEnpCtrl::DumpNetHead(CEnpNetHead& stHead)
{
	CT_TRACE(("%s", "---- BeginDumpNetHead ----"));
	struct in_addr in;
	in.s_addr = stHead.uiRemoteIP;
	CT_TRACE(("%-16s = %s", "RemoteIP", inet_ntoa(in)));
	CT_TRACE(("%-16s = %u", "RemotePort", stHead.unRemotePort));
	CT_TRACE(("%-16s = %u", "Handle", stHead.uiHandle));
	CT_TRACE(("%-16s = %u", "HandleSeq", stHead.uiHandleSeq));
	CT_TRACE(("%-16s = %u", "CreateTime", stHead.uiCreateTime));
	CT_TRACE(("%-16s = %u", "LastTime", stHead.uiLastTime));
	CT_TRACE(("%-16s = %llu", "Reserve1", stHead.u64Act1));
	CT_TRACE(("%-16s = %llu", "Reserve2", stHead.u64Act2));
	CT_TRACE(("%-16s = %u", "LiveFlag", stHead.unLiveFlag));
	CT_TRACE(("%-16s = %u", "DataLength", stHead.unDataLength));
	CT_TRACE(("%s", "---- EndDumpNetHead ----"));
}

void GYEnpCtrl::DumpHandleInfo(GYEnpClient* pstClient)
{
	CT_TRACE(("%s", "---- BeginDumpHandleInfo ----"));
	CT_TRACE(("%-16s = %u", "Handle", pstClient->m_stNetHead.uiHandleSeq));
	struct in_addr in;
	in.s_addr = pstClient->m_stNetHead.uiRemoteIP;
	CT_TRACE(("%-16s = %s", "RemoteIP", inet_ntoa(in)));
	CT_TRACE(("%-16s = %u", "RemotePort", pstClient->m_stNetHead.unRemotePort));
	CT_TRACE(("%-16s = %u", "Socket", pstClient->m_stSocket.GetSocket()));
	CT_TRACE(("%-16s = %u", "HandleSeq", pstClient->m_stNetHead.uiHandleSeq));
	CT_TRACE(("%-16s = %u", "CreateTime", pstClient->m_stNetHead.uiCreateTime));
	CT_TRACE(("%-16s = %u", "LastTime", pstClient->m_stNetHead.uiLastTime));
	CT_TRACE(("%-16s = %llu", "Act1", pstClient->m_stNetHead.u64Act1));
	CT_TRACE(("%-16s = %llu", "Act2", pstClient->m_stNetHead.u64Act2));
	CT_TRACE(("%-16s = %u", "LiveFlag", pstClient->m_stNetHead.unLiveFlag));
	CT_TRACE(("%-16s = %u", "RecvBufAct", pstClient->m_stRecvBuf.Act()));
	CT_TRACE(("%-16s = %u", "RecvBufLen", pstClient->m_stRecvBuf.GetUsedLen()));
	CT_TRACE(("%-16s = %u", "SendBufAct", pstClient->m_stSendBuf.Act()));
	CT_TRACE(("%-16s = %u", "SendBufLen", pstClient->m_stSendBuf.GetUsedLen()));
	CT_TRACE(("%s", "---- EndDumpHandleInfo ----"));
}

int GYEnpCtrl::Run()
{
	int iRet = 0;
	int iLoop = 0;

	m_uiNow = static_cast<unsigned int>(time(0));
	int iPrevNow = m_uiNow;
	while(true)
	{
		// 定时器检查
		if (iLoop == 0)
		{
			m_uiNow = (unsigned int)time(0);
			if (m_uiNow - iPrevNow >= 60)
			{
				iPrevNow = m_uiNow;
				DumpStatInfo();
			}
		}

		iRet = m_stEpoll.WaitAndEvent(10);
		if(iRet == 0)
		{
			//CT_INFO(("m_stEpoll timeout"));
			//==========
		}

		if (++iLoop >= 50)
		{
			iLoop = 0;
		}
	}
	return 0;
}

int GYEnpCtrl::Exit()
{
	CT_INFO(("exit"));
	return 0;
}
